package com.cinyi.cyds.base.mongo;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import com.cinyi.cyds.util.ConfigUtil;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MapReduceCommand;
import com.mongodb.MapReduceOutput;
import com.mongodb.Mongo;
import com.mongodb.WriteResult;


/**
 * MongoDB操作类
 * @author dyk
 *
 */
public class MongoHelper implements IMongoService{
    private final static Logger logger = Logger.getLogger(MongoHelper.class);
    
    private static String DBNAME = "senyintdb";
    private Mongo mongo = null;
    private DB _db = null;
    private String mongoServerAddr;
    private int mongoServerPort;
    private boolean enabled = false ;
    
    private static Map<String, DBCollection> dbCollectionMap = new ConcurrentHashMap<String, DBCollection>();
    
    protected MongoHelper(){
    	mongoServerAddr = ConfigUtil.getString("mongo.pool.host", "mongodb-server");
    	mongoServerPort = ConfigUtil.getInt("mongo.pool.port", 27017);
    	DBNAME = ConfigUtil.getString("mongo.pool.dbname", DBNAME);
    	logger.debug("mongo connect: " + this.mongoServerAddr + ", port: " + this.mongoServerPort + ",database:" + DBNAME);
    	init();
    }
    
    private void init() {
        if (this.mongo == null) {
            try {
                this.mongo = new Mongo( this.mongoServerAddr , this.mongoServerPort);
                if (null != this.mongo) {
                    this._db = this.mongo.getDB(DBNAME);
                    
                    logger.debug("mongo.slave:" + ConfigUtil.getInt("mongo.slave", 0) );
                	//只从secondary中读，如果secondary访问不了的时候就不能进行查询
                	if(ConfigUtil.getInt("mongo.slave", 0) == 1 ){
                		_db.slaveOk();
//                		dbConnection.setReadPreference(ReadPreference.secondary());
	            	}
                    enabled = true ;
                }
            } catch (UnknownHostException e) {
                logger.error("连接mongoDb失败, 服务器地址: " + this.mongoServerAddr + ", 端口: " + this.mongoServerPort);
                throw new RuntimeException(e);
            }
        }
    }
    
    
    
    /**
     * get an table 
     * @param collectionName
     * @return
     */
    private DBCollection getDBCollection(String collectionName) {
        DBCollection collection = null;
        if (dbCollectionMap.containsKey(collectionName)) {
            collection = dbCollectionMap.get(collectionName);
        } else {
            collection = this._db.getCollection(collectionName);
            if (null != collection) {
                dbCollectionMap.put(collectionName, collection);
            }
        }
        return collection;
    }
    
    /**
     * check if doc exsit
     * @param collectionName table name
     * @param query target document
     */
    @Override
    public boolean isDocumentExsit(String collectionName, DBObject query) {
        boolean result = false;
        DBCursor dbCursor = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            dbCursor = collection.find(query);
            if (null != dbCursor && dbCursor.hasNext()) {
                result = true;
            }
        }
        
//        ObjectId id = ObjectId.class.cast(query.get("_id"));
        
        return result;
    }
    /**
     * query an record
     * @param collectionName table name
     * @param query target document
     * @return
     */
    @Override
    public DBObject selectDocument(String collectionName, DBObject query) {
        DBObject result = null;
        DBCursor dbCursor = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            dbCursor = collection.find(query);
            if (null != dbCursor && dbCursor.hasNext()) {
                result = dbCursor.next();
            }
        }
        return result;
    }

    @Override
    public List<DBObject> selectDocuments(String collectionName, DBObject query) {
    	return selectDocuments(collectionName,query,0,20);
    }
    
    @Override
    public List<DBObject> selectDocuments(String collectionName, DBObject query,int skip,int limit) {
        List<DBObject> result = null;
        DBCursor dbCursor = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
        	if(limit>0){
        		if(skip > 0){
        			dbCursor = collection.find(query).skip(skip).limit(limit);
        		}else
        			dbCursor = collection.find(query).limit(limit);
        	}else
        		dbCursor = collection.find(query);
        	
            if (null != dbCursor && dbCursor.size()>0 ) {
            	result = dbCursor.toArray();
            }
        }
        return result;
    }
    
    @Override
    public DBCursor selectDocumentsInCursor(String collectionName, DBObject query,int skip,int limit) {
        DBCursor dbCursor = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
        	if(limit>0){
        		if(skip > 0){
        			dbCursor = collection.find(query).skip(skip).limit(limit);
        		}else
        			dbCursor = collection.find(query).limit(limit);
        	}else
        		dbCursor = collection.find(query);
        	
        }
        return dbCursor;
    }
    
    public Long count(String dbname) {
    	return count(null, dbname) ;
    }
    
    public Long count(DBObject query,String dbname) {
    	DBCollection collection = this.getDBCollection(dbname);
    	
//    	long count=collection.find(queryObject).count();    
//    	long count=collection.count(queryObject);
    	
    	return collection.count(query);
    }
    
    public Long count(String dbname, DBObject query) {
    	DBCollection collection = this.getDBCollection(dbname);
    	long count = collection.getCount(query,new BasicDBObject().append("_id", 1));  
    	
    	return count ;
    }
    
    /**
     *     /**
     * insert an new record
     * @param collectionName table name
     * @param newDocument new doc
     * @param query target document
     */
    @Override
    public void insertDocument(String collectionName, DBObject newDocument) {
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            if (!this.isDocumentExsit(collectionName, newDocument)) {//insert only doc not exist 
                collection.insert(newDocument);
            }
        }
    }
    
    @Override
	public void insertDocuments(String collectionName, List<DBObject> documents) {
    	 DBCollection collection = this.getDBCollection(collectionName);
         if (null != collection) {
             collection.insert(documents);
         }
	}
    
    /**
     * update an document
     * @param collectionName
     * @param query target document
     * @param updatedDocument
     * @return
     */
    @Override
    public boolean updateDocument(String collectionName, DBObject query, DBObject updatedDocument) {
        boolean result = false;
        WriteResult writeResult = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            writeResult = collection.update(query, updatedDocument);
            if (null != writeResult) {
                if (writeResult.getN() > 0) {
                    result = true;
                }
            }
        }
        return result;
    }
    
    @Override
    public boolean updateOrSaveDocument(String collectionName, DBObject query, DBObject updatedDocument, boolean upsert) {
        boolean result = false;
        WriteResult writeResult = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            writeResult = collection.update(query, updatedDocument, upsert, false);
            if (null != writeResult) {
                if (writeResult.getN() > 0) {
                    result = true;
                }
            }
        }
        return result;
    }
    
    /**
     * delete an document
     * @param collectionName
     * @param document target document
     * @return
     */
    @Override
    public boolean deleteDocument(String collectionName, DBObject query) {
        boolean result = false;
        WriteResult writeResult = null;
        DBCollection collection = this.getDBCollection(collectionName);
        if (null != collection) {
            writeResult = collection.remove(query);
            if (null != writeResult) {
                if (writeResult.getN() > 0) {
                    result = true;
                }
            }
        }
        return result;
    }
    
    public DBObject group(String collectionName,DBObject key,DBObject cond,DBObject initial,String reduce,String finalize){
	    try {
	    	DBCollection conn  = this.getDBCollection(collectionName);
	        return conn.group(key, cond, initial, reduce,finalize);
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
		return null;
	}
    
    public DBObject group(MongoGroupModel group){
	    try {
	    	return group(group.getDbName(),group.getKeyDBObject(), group.getCondDBObject(), group.getInitialDBObject(), group.getReduce(),group.getFinalize()) ;
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
		return null;
	}
	
    public List<DBObject> groupByMapCommand(MongoMapCommand mmc){
	    try {
	       DBCollection conn  = this.getDBCollection(mmc.getDbName());
    	   MapReduceCommand cmd = new MapReduceCommand(conn, mmc.getEmit(), mmc.getReduce(),
    			   		null, MapReduceCommand.OutputType.INLINE, mmc.getQueryDBobject());
    	   //MapReduceOutput out = conn.mapReduce(map1,reduce,"message1",null);
    	   if(mmc.getFinalize() != null){
    		   cmd.setFinalize(mmc.getFinalize());
    	   }
    	   MapReduceOutput out = conn.mapReduce(cmd);
    	   return (List<DBObject>) out.results();
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
		return new ArrayList<DBObject>();
	}
    
    public DBCursor groupByMapCommand(MongoMapCommand mmc,String outputCollectionName,int skip,int limit){
    	DBCursor cursor = null; 
    	try {
	       DBCollection conn  = this.getDBCollection(mmc.getDbName());
    	   MapReduceCommand mrc = new MapReduceCommand(conn, mmc.getEmit(),mmc.getReduce(),outputCollectionName,MapReduceCommand.OutputType.REPLACE,mmc.getQueryDBobject());  
    	   if(mmc.getFinalize() != null){
    		   mrc.setFinalize(mmc.getFinalize());
    	   } 
    	   MapReduceOutput out = conn.mapReduce(mrc);  
    	   cursor = out.getOutputCollection().find() ;
    	   if(skip > 0){
    		   cursor.skip(skip);
    	   }
    	   if(limit>0){
    		   cursor.limit(limit); 
    	   }
    	  
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
		return cursor ;
	}
    
    public DBObject groupByAggregate(MongoGroupModel group){
	    try {
	    	return group(group.getDbName(),group.getKeyDBObject(), group.getCondDBObject(), group.getInitialDBObject(), group.getReduce(),group.getFinalize()) ;
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
		return null;
	}
	
    public boolean isEnabled() {
		return enabled;
	}
    
//    public static void main(String[] args) throws UnknownHostException {
//        MongoHelper m = new MongoHelper();
//        String collectionName = "message";
////        String collectionName = "demo";
//        
//        //insert
//        String json ="{'num' : 1}";
//        DBObject doc1 =(DBObject)JSON.parse(json);
//        m.insertDocument(collectionName, doc1);
//        
//        //select
//        DBObject doc2 = null;
//        DBObject query = new BasicDBObject();
//        query.put("num", 1);
//        doc2 = m.selectDocument(collectionName, query);
//        
//        //update
//        DBObject updatedDocument = new BasicDBObject();
//        updatedDocument.put("$set", new BasicDBObject().append("num", 100));
//        boolean result = m.updateDocument(collectionName, query, updatedDocument);
//        query.put("num", 100);
////        //remove
////        result = m.deleteDocument(collectionName, query);
//    }

    public void close(){
    	if(mongo != null ){
    		mongo.close();
    	}
    }
}